package cn.net.yugu.doraemon.datahub.worker;

import cn.net.yugu.doraemon.datahub.config.DataHubProperties;
import cn.net.yugu.doraemon.datahub.config.DataHubConfig;
import cn.net.yugu.doraemon.datahub.core.RecordConverter;
import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.exception.*;
import com.aliyun.datahub.client.model.Field;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.RecordSchema;
import com.aliyun.datahub.clientlibrary.config.ConsumerConfig;
import com.aliyun.datahub.clientlibrary.consumer.Consumer;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @description: data listener worker
 * @author: wanghongli
 * @create: 2022-07-17 11:06:17
 */
@Slf4j
public class RecordConsumeWorkerThread extends Thread {
    private volatile boolean init = false;
    private DataHubConfig dataHubConfig;
    private RecordSchema recordSchema;
    private RecordConverter recordConverter;
    private Class<?> msgClazz;
    private DataHubProperties accountProperties;
    private DatahubClient datahubClient;

    public RecordConsumeWorkerThread(
            DatahubClient datahubClient,
            DataHubProperties accountProperties,
            String workerName,
            DataHubConfig dataHubConfig,
            Class<?> msgClazz) {
        this.datahubClient = datahubClient;
        this.accountProperties = accountProperties;
        this.dataHubConfig = dataHubConfig;
        this.msgClazz = msgClazz;
        this.setName("DataHub-Worker" + workerName);
        this.setDaemon(true);
    }

    @Override
    public void run() {
        initRecordSchema();
        Consumer consumer = null;
        try {
            consumer = createConsumer(accountProperties, dataHubConfig);
            for (;;) {
                if (interrupted()) {
                    log.info("daemon thread {}-{} interrupted", this.getName(), this.getId());
                    break;
                }
                try {
                    RecordEntry recordEntry = consumer.read(0);
                    if (recordEntry == null) {
                        TimeUnit.SECONDS.sleep(1);
                        continue;
                    }
                    Object msg = this.recordConverter.convert2Object(recordEntry);
                    RecordConsumeWorker.invokeMethod(dataHubConfig, msg);
                    recordEntry.getKey().ack();
                } catch (SubscriptionOffsetResetException e) {
                    //非法游标或游标已过期，建议重新定位后开始消费
                    consumer = createConsumer(accountProperties, dataHubConfig);
                    log.error("dataHub:consume msg error happened{}, project:{} topic{}",
                            e.getErrorMessage(), dataHubConfig.getProjectName(), dataHubConfig.getTopicName());
                } catch (InvalidParameterException |
                        SubscriptionOfflineException |
                        SubscriptionSessionInvalidException |
                        AuthorizationFailureException |
                        NoPermissionException e) {
                    log.error("dataHub:consume msg error happened{}, project:{} topic{}",
                            e.getErrorMessage(), dataHubConfig.getProjectName(), dataHubConfig.getTopicName());
                    throw e;
                } catch (DatahubClientException e) {
                    TimeUnit.SECONDS.sleep(1);
                } catch (Exception e) {
                    this.interrupt();
                    log.error("dataHub:consume records exception:{} project:{} topic:{}",
                            e, dataHubConfig.getProjectName(), dataHubConfig.getTopicName());
                }
            }
        } catch (Throwable e) {
            log.error("dataHub:consume records exception:{} project:{} topic:{}", e, dataHubConfig.getProjectName(), dataHubConfig.getTopicName());
        } finally {
            if (consumer != null) {
                consumer.close();
            }
        }
    }

    public static Consumer createConsumer(DataHubProperties accountProperties, DataHubConfig dataHubConfig) {
        ConsumerConfig consumerConfig = new ConsumerConfig(
                accountProperties.getEndpoint(),
                accountProperties.getAccessId(),
                accountProperties.getAccessKey());
        consumerConfig.setAutoCommit(false);
        return new Consumer(
                dataHubConfig.getProjectName(),
                dataHubConfig.getTopicName(),
                dataHubConfig.getSubId(),
                consumerConfig);
    }

    /**
     * 终止
     */
    public void shutdown() {
        if (!interrupted()) {
            interrupt();
        }
    }

    /**
     * 初始化topic字段以及recordSchema
     */
    private void initRecordSchema() {
        try {
            if (!init) {
                recordSchema = this.datahubClient.getTopic(dataHubConfig.getProjectName(),
                        dataHubConfig.getTopicName()).getRecordSchema();
                List<Field> fields = recordSchema.getFields();
                this.recordConverter = new RecordConverter(fields, this.msgClazz);
                init = true;
            }
        } catch (Exception e) {
            log.error("dataHub:init record schema exception:{} project:{} topic:{}",
                    e, dataHubConfig.getProjectName(), dataHubConfig.getTopicName());
        }
    }
}
